//! 配置广播集成功能使用示例

use rustcloud_config::{
    ConfigBroadcastManager, ConfigBroadcastListener, ConfigBroadcastEvent,
    BusConfigBroadcastManager, ConfigResult,
};
use rustcloud_bus::InMemoryMessageBus;
use async_trait::async_trait;
use std::sync::Arc;
use tracing::{info, error};

/// 示例配置变更监听器
struct ExampleConfigListener {
    name: String,
    interested_apps: Vec<String>,
}

impl ExampleConfigListener {
    fn new(name: &str, interested_apps: Vec<String>) -> Self {
        Self {
            name: name.to_string(),
            interested_apps,
        }
    }
}

#[async_trait]
impl ConfigBroadcastListener for ExampleConfigListener {
    async fn on_config_change(&self, event: &ConfigBroadcastEvent) -> ConfigResult<()> {
        info!(
            "监听器 {} 收到配置变更事件: {} -> {} ({})",
            self.name,
            event.event_type.as_str(),
            event.application,
            event.id
        );
        
        match event.event_type {
            rustcloud_config::ConfigBroadcastEventType::ConfigUpdate => {
                if let (Some(path), Some(new_value)) = (&event.config_path, &event.new_value) {
                    info!("配置更新: {} = {}", path, new_value);
                    // 这里可以重新加载配置
                    self.reload_config(&event.application, &event.profile).await?;
                }
            }
            rustcloud_config::ConfigBroadcastEventType::ConfigRefresh => {
                info!("收到配置刷新请求");
                self.refresh_config(&event.application, &event.profile).await?;
            }
            rustcloud_config::ConfigBroadcastEventType::ServiceRestart => {
                info!("收到服务重启请求");
                self.handle_restart_request().await?;
            }
            _ => {
                info!("收到其他类型事件: {:?}", event.event_type);
            }
        }
        
        Ok(())
    }
    
    fn name(&self) -> &str {
        &self.name
    }
    
    fn should_handle(&self, event: &ConfigBroadcastEvent) -> bool {
        // 只处理感兴趣的应用
        if !self.interested_apps.is_empty() {
            self.interested_apps.contains(&event.application)
        } else {
            true
        }
    }
    
    fn interested_applications(&self) -> Vec<String> {
        self.interested_apps.clone()
    }
}

impl ExampleConfigListener {
    /// 重新加载配置
    async fn reload_config(&self, application: &str, profile: &str) -> ConfigResult<()> {
        info!("重新加载配置: {} - {}", application, profile);
        // 实际的配置重新加载逻辑
        Ok(())
    }
    
    /// 刷新配置
    async fn refresh_config(&self, application: &str, profile: &str) -> ConfigResult<()> {
        info!("刷新配置: {} - {}", application, profile);
        // 实际的配置刷新逻辑
        Ok(())
    }
    
    /// 处理重启请求
    async fn handle_restart_request(&self) -> ConfigResult<()> {
        info!("处理服务重启请求");
        // 这里可以实现优雅重启逻辑
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 初始化日志
    tracing_subscriber::init();
    
    info!("🚀 启动配置广播集成示例");
    
    // 1. 创建消息总线
    let bus = Arc::new(InMemoryMessageBus::new().await?);
    info!("✅ 消息总线创建成功");
    
    // 2. 创建配置广播管理器
    let broadcast_manager = BusConfigBroadcastManager::new(
        bus.clone(),
        "config-service".to_string(),
        Some("config.broadcast".to_string()),
    );
    info!("✅ 配置广播管理器创建成功");
    
    // 3. 创建监听器
    let listener1 = Box::new(ExampleConfigListener::new(
        "user-service-listener",
        vec!["user-service".to_string(), "order-service".to_string()],
    ));
    
    let listener2 = Box::new(ExampleConfigListener::new(
        "global-listener",
        vec![], // 监听所有应用
    ));
    
    // 4. 注册监听器
    broadcast_manager.register_listener(listener1).await;
    broadcast_manager.register_listener(listener2).await;
    info!("✅ 配置变更监听器注册成功");
    
    // 5. 启动广播管理器
    broadcast_manager.start().await?;
    info!("✅ 配置广播管理器已启动");
    
    // 6. 模拟配置变更事件
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
    
    info!("📡 发送配置更新事件");
    broadcast_manager.broadcast_config_update(
        "user-service",
        "prod",
        Some("v1.0"),
        "database.url",
        Some(serde_json::Value::String("old-db-url".to_string())),
        serde_json::Value::String("new-db-url".to_string()),
        Some("admin"),
    ).await?;
    
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
    
    info!("📡 发送配置刷新事件");
    broadcast_manager.broadcast_config_refresh(
        "user-service",
        "prod",
        Some(vec!["user-service-instance-1".to_string()]),
    ).await?;
    
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
    
    info!("📡 发送服务重启事件");
    broadcast_manager.broadcast_service_restart(
        "user-service",
        Some(vec!["user-service-instance-1".to_string(), "user-service-instance-2".to_string()]),
        Some("admin"),
    ).await?;
    
    // 7. 等待事件处理
    tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
    
    // 8. 查看统计信息
    let stats = broadcast_manager.get_broadcast_stats().await;
    info!("📊 广播统计信息:");
    info!("  - 发送事件数: {}", stats.events_sent);
    info!("  - 接收事件数: {}", stats.events_received);
    info!("  - 处理成功数: {}", stats.events_processed);
    info!("  - 处理失败数: {}", stats.events_failed);
    info!("  - 活跃监听器: {}", stats.active_listeners);
    
    // 9. 查看集群状态
    let cluster_state = broadcast_manager.get_cluster_state().await;
    info!("🌐 集群状态:");
    for (service, state) in cluster_state {
        info!("  - 服务: {}, 状态: {:?}, 最后心跳: {}", 
              service, state.status, state.last_heartbeat);
    }
    
    // 10. 停止广播管理器
    broadcast_manager.stop().await?;
    info!("🛑 配置广播管理器已停止");
    
    info!("✨ 配置广播集成示例完成");
    
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[tokio::test]
    async fn test_config_broadcast_integration() {
        // 创建测试环境
        let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
        let manager = BusConfigBroadcastManager::new(
            bus,
            "test-service".to_string(),
            None,
        );
        
        // 创建测试监听器
        let listener = Box::new(ExampleConfigListener::new(
            "test-listener",
            vec!["test-app".to_string()],
        ));
        
        // 注册监听器
        manager.register_listener(listener).await;
        
        // 启动管理器
        manager.start().await.unwrap();
        
        // 发送测试事件
        let result = manager.broadcast_config_update(
            "test-app",
            "test",
            None,
            "test.key",
            None,
            serde_json::Value::String("test-value".to_string()),
            None,
        ).await;
        
        assert!(result.is_ok());
        
        // 等待事件处理
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
        
        // 检查统计信息
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 1);
        assert_eq!(stats.active_listeners, 1);
        
        // 停止管理器
        manager.stop().await.unwrap();
    }
    
    #[tokio::test]
    async fn test_listener_filtering() {
        let listener = ExampleConfigListener::new(
            "test-listener",
            vec!["user-service".to_string()],
        );
        
        // 创建测试事件
        let event1 = rustcloud_config::ConfigBroadcastEvent::config_update(
            "user-service",
            "prod",
            None,
            "test.key",
            None,
            serde_json::Value::String("test".to_string()),
            "config-server",
            None,
        );
        
        let event2 = rustcloud_config::ConfigBroadcastEvent::config_update(
            "order-service",
            "prod",
            None,
            "test.key",
            None,
            serde_json::Value::String("test".to_string()),
            "config-server",
            None,
        );
        
        // 测试过滤逻辑
        assert!(listener.should_handle(&event1));
        assert!(!listener.should_handle(&event2));
    }
}